'''
1、接受tessng向kafka发送的车辆数据,并进行展示


'''
import copy
import os
from queue import Queue
from concurrent.futures import ThreadPoolExecutor
import carla
import numpy as np
import time
import redis
import random
from carla import Transform, Rotation, Location
from kafka import KafkaConsumer, KafkaProducer
import json
from pyproj import Proj


class myCarla:
    def __init__(self, carla_ip, carla_port):

        # ==================carla初始化======================
        # 创建carla客户端
        self.client = carla.Client(carla_ip, carla_port)
        self.client.set_timeout(10.0)
        self.world = self.client.load_world('tongji')
        self.settings = self.world.get_settings()
        # 获取carla中的交通管理者
        self.traffic_manager = self.client.get_trafficmanager(8000)
        # ==================基本信息======================
        # 转换坐标
        self.rule = Proj("+proj=tmerc +lon_0=121.2092870660126 +lat_0=31.292829882838856 +ellps=WGS84")

        # ==================缓存======================
        # carla中运行的主车，key-已经生成的车辆id，value-carla中id
        self.run_bjc_carla_id_dict = {}
        # carla中运行的背景车，key-已经生成的车辆id，value-carla中id
        self.run_target_carla_id_dict = {}
        self.now_car_type_carla_type_dist = {
            0: 'vehicle.ford.crown',  # 黄色小客车【保守主车】
            1: 'vehicle.seat.leon',  # 蓝色小客车【激进主车】
            2: 'vehicle.mini.cooper_s_2021',  # 红色小客车
            3: 'vehicle.nissan.patrol_2021',  # 白色小客车
            4: 'vehicle.jeep.wrangler_rubicon',  # 灰色小客车
            5: 'vehicle.lincoln.mkz_2017'  # 黑色小客车
        }
        # carla车辆类型缓存
        # 默认车辆类型
        self.default_carla_type = 'vehicle.lincoln.mkz_2017'

        # ==================雷达点云、相机======================
        # 【相机图片、雷达点云】数据保存路径
        self.root_path = r'D:\hsy\tessng\tessng-project\tongji\carla'
        # 存放雷达点云数据
        self.sensor_queue = Queue()
        # 存放 绑定后的雷达
        self.sensor_list = []
        # 雷达点云展示
        self.lidar_point_cloud_show = True
        # 雷达点云存储
        self.lidar_point_cloud_save = False
        self.to_reset_view_point = True
        self.vis = None
        # ==================kafka======================
        self.bootstrap_servers = '10.102.1.164:9092'
        # # 初始化 Kafka 生产者
        # self.kafka_producer = KafkaProducer(
        #     bootstrap_servers=self.bootstrap_servers,
        #     value_serializer=lambda v: json.dumps(v).encode('utf-8')  # 将数据序列化为 JSON
        # )
        # ==================redis======================
        # self.redis_host = '10.100.12.90'
        # self.redis_port = 6379
        # self.redis_password = 'Wanji@300552!'
        # self.redis_client = redis.Redis(host=self.redis_host, port=self.redis_port, password=self.redis_password, db=0)

    # 设置天气
    def set_weather(self):
        weather = carla.WeatherParameters(
            cloudiness=10.0,  # 0-100  0 是晴朗的天空，100 是完全阴天
            precipitation=0.0,  # 0 表示没有下雨，100 表示大雨
            precipitation_deposits=0.0,  # 0 表示道路上没有水坑，100 表示道路完全被雨水覆盖
            wind_intensity=0.0,  # 0 表示平静，100 表示强风，风会影响雨向和树叶
            sun_azimuth_angle=0.0,  # 太阳方位角，0～360
            sun_altitude_angle=90.0,  # 太阳高度角，90 是中午，-90 是午夜
            fog_density=0.0,  # 0～100 表示雾的浓度或厚度，仅影响RGB相机传感器
            fog_distance=0.0,  # 雾开始的距离，单位为米
            wetness=0.0,  # 0～100 表示道路湿度百分比，仅影响RGB相机传感器
            fog_falloff=0.0,  # 雾的密度，0至无穷大，0 表示雾比空气轻，覆盖整个场景，1表示与空气一样，覆盖正常大小的建筑物
            scattering_intensity=0.0,  # 控制光线对雾的穿透程度
            mie_scattering_scale=0.0,  # 控制光线与花粉或空气等大颗粒的相互作用，导致天气朦胧，光源周围有光晕，0表示无影响
            rayleigh_scattering_scale=0.0331,  # 控制光与空气分子等小粒子的相互作用，取决于光波长，导致白天蓝天或晚上红天
        )
        self.world.set_weather(weather)

    # 设置视角坐标
    def set_view(self, lon, lat):
        x, y = self.rule(lon, lat)
        z = 215
        view_angle = Transform(Location(x=-707.612793, y=161.383148, z=246.703506),
                               Rotation(pitch=-88.999451, yaw=2.327019, roll=-133.665176))
        self.world.get_spectator().set_transform(view_angle)  # 设置世界视角

    # 创建交通管理器对象
    def set_traffic_manager(self):
        self.traffic_manager.set_global_distance_to_leading_vehicle(2.0)
        self.traffic_manager.global_percentage_speed_difference(40.0)
        self.traffic_manager.set_synchronous_mode(True)
        if not self.settings.synchronous_mode:
            print("开启异步模式")
            synchronous_master = True
            self.settings.synchronous_mode = False
            self.settings.fixed_delta_seconds = 0.1
            self.world.apply_settings(self.settings)

    # 获取carla车辆
    def get_carla_car_type(self, veh_type):
        if veh_type in self.now_car_type_carla_type_dist.keys():
            carla_type = self.now_car_type_carla_type_dist[veh_type]
        else:
            carla_type = self.default_carla_type
        carla_car = self.world.get_blueprint_library().find(carla_type)
        # carla_car.set_simulate_physics(True)  # 启用物理模拟（恢复碰撞检测）
        return carla_car

    # 获取carla车辆信息
    @staticmethod
    def get_vehicle_info(vehicle):
        transform = vehicle.get_transform()
        velocity = vehicle.get_velocity()
        return {
            'id': vehicle.id,
            'type_id': vehicle.type_id,
            'location': {
                'x': transform.location.x,
                'y': transform.location.y,
                'z': transform.location.z
            },
            'rotation': {
                'pitch': transform.rotation.pitch,
                'yaw': transform.rotation.yaw,
                'roll': transform.rotation.roll
            },
            'velocity': {
                'x': velocity.x,
                'y': velocity.y,
                'z': velocity.z
            }
        }

    # 通过redis，获取车辆信息并设置
    def set_car_for_redis(self, redis_channel):
        pubsub = self.redis_client.pubsub()
        pubsub.subscribe(redis_channel)
        try:
            while True:
                for message in pubsub.listen():
                    if message['type'] == 'message':
                        # 接收到数据，添加处理逻辑
                        receive_data = message['data'].decode('utf-8')
                        data = json.loads(receive_data)
                        results = data['value']['value']
                        # 当前帧创建失败的车辆id集合
                        now_error_id_list = []
                        # 当前帧所有车辆id集合
                        now_id_list = []
                        for p in results:
                            now_id_list.append(p["id"])
                        # 添加交通流
                        for par in results:
                            x, y = self.rule(par["longitude"], par["latitude"])
                            veh_type = par['vehicleType']
                            car_id = par['id']
                            angle = par["courseAngle"]
                            # 判断是否已经生成过该车辆
                            if car_id not in self.run_bjc_carla_id_dict.keys():
                                # 没有生成过，则生成
                                trans = Transform(carla.Location(x=x, y=-y, z=15), carla.Rotation(yaw=angle - 90))
                                # 根据车辆类型生成车辆
                                if car_id in self.target_id:
                                    # 主车
                                    carla_car = self.get_carla_car_type(0)
                                else:
                                    # 背景车
                                    carla_car = self.get_carla_car_type(veh_type)
                                # 将车辆生成请求添加到批量请求中
                                batch = [
                                    carla.command.SpawnActor(carla_car, trans).then(
                                        carla.command.SetSimulatePhysics(carla.command.FutureActor, False))
                                ]
                                # 将车辆生成请求发送到服务器端，并获取响应
                                response = self.client.apply_batch_sync(batch, False)[0]
                                if response.error:
                                    now_error_id_list.append(car_id)
                                else:
                                    self.run_bjc_carla_id_dict[car_id] = response.actor_id
                            else:
                                # 生成过，则更新位置和角度
                                # 更新位置和角度
                                trans = Transform(carla.Location(x=x, y=-y, z=15),
                                                  carla.Rotation(yaw=par["courseAngle"] - 90))
                                carla_id = self.run_bjc_carla_id_dict[car_id]
                                carla_car = self.world.get_actor(carla_id)
                                if carla_car is not None:
                                    carla_car.set_transform(trans)
                        print(f'当前carla中运行车辆个数：{len(self.run_bjc_carla_id_dict)}')
                        # 销毁不在当前帧的id
                        destroy_id_list = [id for id in self.run_bjc_carla_id_dict if id not in now_id_list]
                        for destroy_id in destroy_id_list:
                            carla_id = self.run_bjc_carla_id_dict[destroy_id]
                            vehicle = self.world.get_actor(carla_id)
                            if vehicle is not None:
                                vehicle.destroy()
                            del self.run_bjc_carla_id_dict[destroy_id]
                        print(f'本次更新数量：{len(now_id_list)}，销毁数量：{len(destroy_id_list)}')

        finally:
            print('开始删除所有创建的车辆')
            # 删除所有动态生成的 Actor
            actors = myCarla.world.get_actors()
            for actor in actors:
                # 删除所有车辆
                if actor.type_id.startswith('vehicle.'):
                    print(f"Destroying vehicle: {actor.id} ({actor.type_id})")
                    actor.destroy()

    # 通过kafka，获取车辆信息并设置
    def set_car_for_kafka(self, topic):
        consumer = KafkaConsumer(
            topic,
            bootstrap_servers=self.bootstrap_servers,
            auto_offset_reset='latest',
            enable_auto_commit=False
        )
        try:
            for message in consumer:
                if message:
                    value = message.value.decode('utf-8')
                    json_data = json.loads(value)
                    results = json_data['value']['value']
                    now_error_id_list = []
                    # 当前帧所有车辆id集合
                    now_id_list = []
                    for p in results:
                        now_id_list.append(p["id"])
                    # 添加交通流
                    for par in results:
                        x, y = self.rule(par["longitude"], par["latitude"])
                        veh_type = par['vehicleType']
                        car_id = par['id']
                        drive_type = par['driveType']
                        angle = par["courseAngle"]
                        # 判断是否已经生成过该车辆
                        if car_id not in self.run_bjc_carla_id_dict.keys():
                            # 没有生成过，则生成
                            trans = Transform(carla.Location(x=x, y=-y, z=15), carla.Rotation(yaw=angle - 90))
                            # 根据车辆类型生成车辆
                            if drive_type == 1:
                                # 主车
                                if car_id in ['4', '5']:
                                    carla_car = self.get_carla_car_type(1)
                                else:
                                    carla_car = self.get_carla_car_type(0)
                            else:
                                carla_type = random.choice([2, 3, 4, 5])
                                carla_car = self.get_carla_car_type(carla_type)
                            # 将车辆生成请求添加到批量请求中
                            batch = [
                                carla.command.SpawnActor(carla_car, trans).then(
                                    carla.command.SetSimulatePhysics(carla.command.FutureActor, False))
                            ]
                            # 将车辆生成请求发送到服务器端，并获取响应
                            response = self.client.apply_batch_sync(batch, False)[0]
                            if response.error:
                                now_error_id_list.append(car_id)
                            else:
                                self.run_bjc_carla_id_dict[car_id] = response.actor_id
                        else:
                            # 生成过，则更新位置和角度

                            carla_id = self.run_bjc_carla_id_dict[car_id]
                            carla_car = self.world.get_actor(carla_id)
                            if carla_car is not None:
                                location = carla_car.get_location()
                                z = location.z
                                # 更新位置和角度
                                trans = Transform(carla.Location(x=x, y=-y, z=z),
                                                  carla.Rotation(yaw=angle - 90))
                                carla_car.set_transform(trans)
                    print(f'当前carla中运行车辆个数：{len(self.run_bjc_carla_id_dict)}')
                    # 销毁不在当前帧的id
                    destroy_id_list = [id for id in self.run_bjc_carla_id_dict if id not in now_id_list]
                    for destroy_id in destroy_id_list:
                        carla_id = self.run_bjc_carla_id_dict[destroy_id]
                        vehicle = self.world.get_actor(carla_id)
                        if vehicle is not None:
                            vehicle.destroy()
                        del self.run_bjc_carla_id_dict[destroy_id]
                    print(f'本次更新数量：{len(now_id_list)}，销毁数量：{len(destroy_id_list)}')

        finally:
            print('开始删除所有创建的车辆')
            # 删除所有动态生成的 Actor
            actors = myCarla.world.get_actors()
            for actor in actors:
                # 删除所有车辆
                if actor.type_id.startswith('vehicle.'):
                    print(f"Destroying vehicle: {actor.id} ({actor.type_id})")
                    actor.destroy()

    # 发送数据到kafka
    def send_data_for_kafka(self, send_topic):
        while True:
            # 获取场景中的所有车辆
            carla_veh_list = self.world.get_actors().filter('vehicle.*')
            vehicle_list = []
            # 遍历所有车辆并获取信息
            for vehicle in carla_veh_list:
                vehicle_info = self.get_vehicle_info(vehicle)
                vehicle_list.append(vehicle_info)
            # print(vehicle_list)
            self.kafka_producer.send(topic=send_topic, value=vehicle_list, key='key')
            # 休眠一段时间，避免频繁发送
            time.sleep(1)

    def get_spectator(self):
        while True:
            # 获取 Spectator
            spectator = self.world.get_spectator()
            # 获取 Spectator 的 Transform
            spectator_transform = spectator.get_transform()
            # 提取位置和旋转信息
            location = spectator_transform.location
            rotation = spectator_transform.rotation
            print(f"Spectator Location: {location}")
            print(f"Spectator Rotation: {rotation}")
            time.sleep(2)


if __name__ == "__main__":
    # myCarla = myCarla('10.100.16.98', 2000)
    myCarla = myCarla('127.0.0.1', 2000)
    executor = ThreadPoolExecutor(max_workers=3)
    myCarla.set_weather()
    myCarla.set_view(121.20237823425242, 31.291840026331244)
    myCarla.set_traffic_manager()
    # 接收并设置背景车辆信息
    set_bj_car_topic = 'bjc_channel'
    executor.submit(myCarla.set_car_for_kafka, set_bj_car_topic)
    # executor.submit(myCarla.set_car_for_redis, set_bj_car_topic)
    # executor.submit(myCarla.get_spectator)
